def test_connection():
conn = connect_db()
cursor = conn.cursor()
for query in (RADCOM,STEL):
cursor.execute(query)
test_eq(type(cursor.fetchone()), pyodbc.Row)Atualização
Este módulo atualiza as bases. Executa as queries sql do STEL, RADCOM e baixa os arquivos de estações e plano básico do MOSAICO.
Conexão com o banco de dados
A função a seguir é um wrapper simples que utiliza o pyodbc para se conectar ao banco de dados base da Anatel e retorna o objeto da conexão
connect_db
connect_db (server:str='ANATELBDRO05', database:str='SITARWEB', trusted_conn:str='yes', mult_results:bool=True)
Conecta ao Banco server e retorna o ‘cursor’ (iterador) do Banco
| Type | Default | Details | |
|---|---|---|---|
| server | str | ANATELBDRO05 | Servidor do Banco de Dados |
| database | str | SITARWEB | Nome do Banco de Dados |
| trusted_conn | str | yes | Conexão Segura: yes | no |
| mult_results | bool | True | Múltiplos Resultados |
| Returns | Connection |
clean_mosaico
clean_mosaico (df:pandas.core.frame.DataFrame, pasta:Union[str,pathlib.Path])
Clean the merged dataframe with the data from the MOSAICO page
| Type | Details | |
|---|---|---|
| df | DataFrame | DataFrame com os dados de Estações e Plano_Básico mesclados |
| pasta | Union | Pasta com os dados de municípios para imputar coordenadas ausentes |
| Returns | DataFrame | DataFrame com os dados mesclados e limpos |
Atualização das bases de dados
As bases de dados são atualizadas atráves das funções a seguir, o único argumento passado em todas elas é a pasta na qual os arquivos locais processados serão salvos, os nomes dos arquivos são padronizados e não podem ser editados para que as funções de leitura e processamento recebam somente a pasta na qual esses arquivos foram salvos.
update_radcom
update_radcom (folder:Union[str,pathlib.Path])
Atualiza a tabela local retornada pela query RADCOM
| Type | Details | |
|---|---|---|
| folder | Union | Pasta onde salvar os arquivos |
| Returns | DataFrame | DataFrame com os dados atualizados |
update_stel
update_stel (folder:Union[str,pathlib.Path])
Atualiza a tabela local retornada pela query STEL
| Type | Details | |
|---|---|---|
| folder | Union | Pasta onde salvar os arquivos |
| Returns | DataFrame | DataFrame com os dados atualizados |
update_mosaico
update_mosaico (folder:Union[str,pathlib.Path])
Atualiza a tabela local do Mosaico. É baixado e processado arquivos xml zipados da página pública do Spectrum E
| Type | Details | |
|---|---|---|
| folder | Union | Pasta onde salvar os arquivos |
| Returns | DataFrame | DataFrame com os dados atualizados |
update_base
update_base (folder:Union[str,pathlib.Path])
Wrapper que atualiza opcionalmente lê e atualiza as três bases indicadas anteriormente, as combina e salva o arquivo consolidado na folder folder
| Type | Details | |
|---|---|---|
| folder | Union | Pasta onde salvar os arquivos |
| Returns | DataFrame | DataFrame com os dados atualizados |
# from urllib.request import Request, urlopen
# from urllib.error import URLError
# req = Request(ESTACAO)
# try:
# response = urlopen(req)
# except URLError as e:
# if hasattr(e, 'reason'):
# print('We failed to reach a server.')
# print('Reason: ', e.reason)
# elif hasattr(e, 'code'):
# print('The server couldn\'t fulfill the request.')
# print('Error code: ', e.code)
# else:
# Path.cwd().joinpath('estações.zip').write_bytes(response.read())